package com.atguigu.flink.sql.catalog;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

import java.util.Arrays;

/**
 * Created by 黄凯 on 2023/6/28 0028 10:51
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 *
 * Catalog: 目录。
 *  *
 *  *   Mysql:
 *  *      test
 *  *         user
 *  *   Oracle:
 *  *      test
 *  *        user
 *  *  [目录名].[库名].[表名]
 */
public class Flink01_HiveCataLog {

    public static void main(String[] args) {

        System.setProperty("HADOOP_USER_NAME","atguigu");

        TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());

        //创建HiveCatalog
        HiveCatalog hiveCatalog = new HiveCatalog(
                "hive",
                "default",
                "conf"
        );

        //注册Catalog
        tableEnv.registerCatalog("hive" , hiveCatalog);
        //使用catalog
        tableEnv.useCatalog("hive");

        //查看当前使用的catalog
        // default_catalog， 默认使用的catalog , GenericInMemoryCatalog
        // 这个catalog的特点就是即将表的元数据信息直接维护到内存中。
        String currentCataglog = tableEnv.getCurrentCatalog();
        System.out.println(currentCataglog);

        //查看当前使用的库
        // default_database ,默认使用的database
        String currentDatabase = tableEnv.getCurrentDatabase();
        System.out.println(currentDatabase);

        //查看当前catalog下所有的表
        System.out.println(Arrays.toString(tableEnv.listTables()));

        //直接查询catalog下表的数据
        tableEnv.sqlQuery("select * from t1").execute().print();

        //在HiveCatalog中创建表: 可以。
        //默认情况下创建的表都属于default_catalog.default_database
        /*
        String readSql=
                " create table t3 (" +
                        " id STRING , " +
                        " vc INT , " +
                        " ts BIGINT " +
                        ") WITH ( " +
                        " 'connector' = 'filesystem' , " +
                        " 'path' = 'input/ws.txt', " +
                        " 'format' = 'csv' " +
                        ")" ;
        tableEnv.executeSql(readSql);
 */

        tableEnv.sqlQuery("select id ,vc ,ts from t3").execute().print();

    }

}
